﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

using System.Collections.Concurrent;
using System.Runtime.CompilerServices;

namespace AzureMRCore.DataCommunication
{
    public class ReduceDataReceiver
    {
        private static ReduceDataReceiver _receiver;

        private ConcurrentDictionary<string, ConcurrentDictionary<int, DataItem>> _fetchingReduceData;
        //   ConcurrentDictionary<INKEY, List<INVALUE>>
        // private readonly ConcurrentDictionary<string, IInputFormat> _inMemCache;

        private ReduceDataReceiver()
        {
            _fetchingReduceData = new ConcurrentDictionary<string, ConcurrentDictionary<int, DataItem>>();
            // _inMemCache = new ConcurrentDictionary<string, IInputFormat>();
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public static ReduceDataReceiver GetInstance()
        {
            return _receiver ?? (_receiver = new ReduceDataReceiver());
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public void AddData(DataItem dataItem)
        {
            string key = dataItem.JobID + "_" + dataItem.Iteration + "_" + dataItem.AppName;
            var bCastDataProducts = _fetchingReduceData.GetOrAdd(key,
                                                                 data =>
                                                                 new ConcurrentDictionary<int, DataItem>());
            bCastDataProducts.TryAdd(dataItem.MapIndex, dataItem);
            /*  if (bCastDataProducts.Count == dataItem.NumMapTasks)
            {
                inputFormat.Init(bCastDataProducts, dataItem.NumMapTasks);
                _inMemCache.TryAdd(key, inputFormat);
                ConcurrentDictionary<int, ReduceDataProduct> task;
                _fetchingReduceData.TryRemove(key, out task);

                // trying to remove old broadcast data from memcache.
                string oldKey = dataItem.JobID + "_" + (dataItem.Iteration - 1);
                IInputFormat temp;
                _inMemCache.TryRemove(oldKey, out temp);
            }*/
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public bool ContainsReduceData(string tableName)
        {
            return _fetchingReduceData.ContainsKey(tableName);
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public ConcurrentDictionary<int, DataItem> GetReduceData(string tableName)
        {
            var item = _fetchingReduceData[tableName];
            return item;
        }
    }
}